package boot.spring.utils;

import boot.spring.config.CloudStorageConfig;
import boot.spring.mapper.UserAmountMapper;
import boot.spring.po.UserAmount;
import boot.spring.service.WebSocketServer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.SneakyThrows;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class CanalClient{

    public static CloudStorageConfig cloudStorageConfig;
    public static UserAmountMapper userAmountMapper;
    public static WebSocketServer webSocketServer;
    public static RedisTemplate redisTemplate;
    public static String oldVal;
    public static CanalConnector connector;
    public static int batchSize = 1000;
    public static int times = 0;

    public void start() {
        Thread thread = new StartCanalThread();
        thread.start();
    }

    /**
     * canal 线程
     */
    public static class StartCanalThread extends Thread {
        @SneakyThrows
        @Override
        public void run() {
            try {
                // 标注 canal服务器开启了
                redisTemplate.opsForValue().set("canalServerOpen","1",15, TimeUnit.SECONDS);
                connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), "example", "", "");
                connector.connect();
                connector.subscribe("tp\\..*");
                connector.rollback();
                cycle(connector,batchSize);
            } catch (Exception err){
                // 标注 canal服务器关闭了
                redisTemplate.opsForValue().set("canalServerOpen","0",15, TimeUnit.SECONDS);
                Thread.sleep(10000);
                times++;
                System.out.println("服务器断了，正在尝试连接：" + times);
                run();
            }
        }
    }

    public static void cycle(CanalConnector connector, Integer batchSize){
        System.out.println("连接上了。");
        while (true) {
            Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
            long batchId = message.getId();
            int size = message.getEntries().size();
            if (batchId == -1 || size == 0) {
                try {
                  redisTemplate.opsForValue().set("canalServerOpen","1",15, TimeUnit.SECONDS);
                  Thread.sleep(1000);
                } catch (InterruptedException e) {
                  e.printStackTrace();
                }
            } else {
                printEntry(message.getEntries());
            }
            connector.ack(batchId); // 提交确认
        }
    }

    /**
     * 数据库执行的操作
     * @param entrys
     */
    private static void printEntry( List<Entry> entrys) {
        for (Entry entry : entrys) {
            //操作事物 忽略
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { continue; }
            RowChange rowChage = null;//执行事件信息
            String database=null;//执行的数据库
            String table=null;//执行的表
            try {
                database=entry.getHeader().getSchemaName();
                table=entry.getHeader().getTableName();
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                System.out.println("获取数据失败："+e.getMessage());
            }
            //获取执行的事件
            EventType eventType = rowChage.getEventType();
            for (RowData rowData : rowChage.getRowDatasList()) {
//                //删除操作
//                if (eventType.equals(EventType.DELETE)) {
//                    printColumn(rowData.getBeforeColumnsList(),"DELETE",database,table);
//                    redisDelete(rowData.getBeforeColumnsList(),database,table);
//                    updateData( rowData.getAfterColumnsList(),database,table);
//                }
//                //添加操作
//                else if (eventType.equals(EventType.INSERT)) {
//                    printColumn(rowData.getBeforeColumnsList(),"INSERT",database,table);
//                    redisInsert(rowData.getAfterColumnsList(),database,table);
//                    updateData( rowData.getAfterColumnsList(),database,table);
//                }
//                //修改操作
//                else if(eventType.equals(EventType.UPDATE)) {
//                    printColumn(rowData.getBeforeColumnsList(),"UPDATE",database,table);
//                    redisUpdate(rowData.getAfterColumnsList(),database,table);
//                    updateData( rowData.getAfterColumnsList(),database,table);
//                }
//                //修改表结构
//                else if(eventType.equals(EventType.ALTER)){
//                    System.out.println("修改表结构：");
//                    printColumn(rowData.getBeforeColumnsList(),"ALTER",database,table);
//                    updateData( rowData.getAfterColumnsList(),database,table);
//                }
                printColumn(rowData.getBeforeColumnsList(),eventType.toString(),database,table);
                updateData( rowData.getAfterColumnsList(),database,table);
            }
        }
    }


    private static void printColumn( List<Column> columns,String type,String database,String table) {
        for (Column column : columns) {
            System.out.println(database+":" + table+":"+column.getName() + " : " + column.getValue() + "    "+type+"=");
        }
    }

    private static void updateData( List<Column> columns,String database,String table){
        // 获取最新的用户数据并且推送
        if (cloudStorageConfig.getMyDbName().equals(database) && "user_amount".equals(table)) {
            JSONObject json=new JSONObject();
            for (Column column : columns) {
                json.put(column.getName(), column.getValue());
            }
            if(columns.size()>0) {
                getNewestUserAmount(JSON.toJSONString(json));
            }
        }
    }

    private static void redisInsert( List<Column> columns,String database,String table){
        JSONObject json=new JSONObject();
        for (Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if(columns.size()>0){
            String val = database + ":" + table + ":" + columns.get(0).getValue();
            redisTemplate.opsForValue().set(val,json);
            updateTableUpdateTime(database,table);
        }
    }

    private static  void redisUpdate( List<Column> columns,String database,String table){
        JSONObject json=new JSONObject();
        for (Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if(columns.size()>0){
            String val = database + ":" + table + ":" + columns.get(0).getValue();
            redisTemplate.opsForValue().set(val,json);
            updateTableUpdateTime(database,table);
        }
    }

    private static  void redisDelete( List<Column> columns,String database,String table){
        JSONObject json=new JSONObject();
        for (Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if(columns.size()>0){
            String val = database + ":" + table + ":" + columns.get(0).getValue();
            redisTemplate.delete(val);
            updateTableUpdateTime(database,table);
        }
    }

    public static void updateTableUpdateTime(String database,String table){
        // 保存这个表更新的时间
        redisTemplate.opsForValue().set(database + ":" + table + ":" +"updateTime",new Date().getTime());
    }

    // 获取最新的用户 充值数据
    public static void getNewestUserAmount(String newVal){
        try {
            final List<UserAmount> userAmounts = userAmountMapper.selectUserAmount();
            String val = JSON.toJSONString(userAmounts);
            webSocketServer.broadcast(val);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
